/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.zeromq.ZContext;
import org.zeromq.ZLoop;
import org.zeromq.ZLoop.IZLoopHandler;
import org.zeromq.ZMQ; 
import org.zeromq.ZMsg;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;
import org.apache.log4j.Logger;

public class ZmqDispatcher {
    private ZContext ctx;
    private String conn;
    private int timeout;
    private int retry;
    private ZmqConsumerP iack;
    private int pos;
    
    private Socket client = null;
    private String id;
    private int retriesLeft;
    private ExecutorService executor;
    private byte[] message;
    private byte[] rid;
    
    private static Logger LOG = Logger.getLogger(ZmqDispatcher.class.getName());
    
    public ZmqDispatcher(ZContext ctx, String conn, int timeout, int retry,
    		ZmqConsumerP iack, int pos) {
        this.ctx = ctx;
        this.conn = conn;
        this.timeout = timeout;
        this.retry = retry;
        this.iack = iack;
        this.pos = pos;
        
        connect();
        executor = Executors.newFixedThreadPool(1);
    }

    private void connect() {
    	disconnect();
    	client = ctx.createSocket(ZMQ.DEALER);
    	id = CSBUtils.sockId();
    	client.setIdentity(id.getBytes());
    	client.connect(conn);
        LOG.info(id + ": connected to router via " + conn + "...");
    }
    
    public void disconnect() {
    	if (client != null)
    		ctx.destroySocket(client);
    }
     
    private IZLoopHandler resend = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
        	if (--retriesLeft <= 0) {
        		connect();
        		((ZmqConsumerP)arg).ack(pos, ZmqConsumerBase.TIMEOUT);
        		return -1;
        	}
        	
			LOG.error(id + ": no response from router, resending...");
			send();
            return 0;
        }
    };
    
    private IZLoopHandler proc = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {        	
			ZMsg msg = ZMsg.recvMsg(client);
			if (msg == null) {
				LOG.error(id + ": receiving interrupted, closing...");
        		connect();
        		((ZmqConsumerP)arg).ack(pos, ZmqConsumerBase.INTERRUPTED);
				return -1;    
			}
            
			msg.unwrap();
			String reply = new String(msg.pop().getData());
			if (reply.equals(CSB.WORKER_ACK)) { 
				msg.destroy();                
				((ZmqConsumerP)arg).ack(pos, ZmqConsumerBase.ACKNOWLEDGED); 
				return -1;
			} else if (reply.equals(CSB.ROUTER_UNREACHABLE)) {
				msg.destroy();
				((ZmqConsumerP)arg).ack(pos, ZmqConsumerBase.UNREACHABLE);
				return -1;
			} else {			
				LOG.warn("Malformed reply from router: " + reply);
			}
			return 0;
        }
    };
    
    private void send() {
	    ZMsg msg = new ZMsg();
	    msg.add(message);
	    if (rid != null)
	    	msg.push(id);
	    msg.send(client);
    }
    
    public void dispatch(final byte[] message, byte[] rid) {    	
    	this.message = message;
    	this.rid = rid;
    	
        retriesLeft = retry;
        executor.submit(new Runnable() {        	
        	@Override
        	public void run () { 
        		
        		send();
        	    
        		ZLoop loop = new ZLoop();
        		PollItem poller = new PollItem(client, Poller.POLLIN);
        		loop.addPoller(poller, proc, iack);
        		loop.addTimer(timeout, retry, resend, iack);        		
        		loop.start();
        		loop.destroy();
        	}      
        }); 
    }    
}

